package javaVersion.learn.flinkWindows;

import javaVersion.pojo.Pro1_Window_WordCountCart;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @program: myFlink
 * @description: 滑动窗口, 有重叠数据
 * @author: WincoMa
 * @create: 2020-07-20 16:33
 **/
public class Pro2_SlideWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream = senv.socketTextStream("hadoop1", 9999);
        stream.map(new MapFunction<String, Pro1_Window_WordCountCart>() {
            @Override
            public Pro1_Window_WordCountCart map(String value) throws Exception {
                String[] word = value.split(" ");
                return new Pro1_Window_WordCountCart(Integer.parseInt(word[0]),Integer.parseInt(word[1]));
            }
        })
                .keyBy(0)
                .timeWindow(Time.seconds(10),Time.seconds(2))
                .sum(1)
                .print();
        senv.execute();

    }
}
